package notification

import (
	"fmt"
	"github.com/mjiulee/lego"
	"sync"
	"time"

	"github.com/garyburd/redigo/redis"
)

/*消息和监听者列表结构*/
type NotifyChannel struct {
	Msg     string       // 消息名
	Sublsit []Subscriber // 监听者的列表
}

/*消息注册中心*/
type registCenter struct {
	chanels []*NotifyChannel // 注册中心管理的消息列表
}

var __once sync.Once
var __ntfcenter *registCenter = nil

func init() {
	__once.Do(func() {
		__ntfcenter = new(registCenter)
		__ntfcenter.chanels = make([]*NotifyChannel, 0)
	})
}

/* 向中心注册消息 */
func Regist(reistor Subscriber, msg string) {
	if __ntfcenter == nil {
		lego.LogError("消息中心初始化失败")
		return
	}

	msgcindex := -1
	for i, mc := range __ntfcenter.chanels {
		if mc.Msg == msg {
			msgcindex = i
			break
		}
	}
	if msgcindex >= 0 && msgcindex < len(__ntfcenter.chanels) {
		registbefore := false
		for _, sub := range __ntfcenter.chanels[msgcindex].Sublsit {
			if sub == reistor {
				registbefore = true
				break
			}
		}
		if !registbefore {
			__ntfcenter.chanels[msgcindex].Sublsit = append(__ntfcenter.chanels[msgcindex].Sublsit, reistor)
		}
	} else {
		newc := new(NotifyChannel)
		newc.Msg = msg
		newc.Sublsit = make([]Subscriber, 0)
		newc.Sublsit = append(newc.Sublsit, reistor)
		__ntfcenter.chanels = append(__ntfcenter.chanels, newc)
	}
}

/* 向中心注销自己 */
func UndoRegist(reistor Subscriber, msg string) {
	if __ntfcenter == nil {
		lego.LogError("消息中心初始化失败")
		return
	}

	msgcindex := -1
	for i, mc := range __ntfcenter.chanels {
		if mc.Msg == msg {
			msgcindex = i
			break
		}
	}
	if msgcindex >= 0 && msgcindex < len(__ntfcenter.chanels) {
		registeridx := -1
		for i, sub := range __ntfcenter.chanels[msgcindex].Sublsit {
			if sub == reistor {
				registeridx = i
				break
			}
		}
		if registeridx >= 0 && registeridx < len(__ntfcenter.chanels[msgcindex].Sublsit) {
			__ntfcenter.chanels[msgcindex].Sublsit = append(__ntfcenter.chanels[msgcindex].Sublsit[:registeridx], __ntfcenter.chanels[msgcindex].Sublsit[registeridx+1:]...)
		}
	}
}

/* 发送local消息，给应用各个模块 */
func PostLocalMsg(chanal, msg string) {
	for i, mc := range __ntfcenter.chanels {
		if mc.Msg == chanal {
			for _, sub := range __ntfcenter.chanels[i].Sublsit {
					sub.OnMsgNotify(chanal, msg)
			}
		}
	}
}

/* 发送local消息，给应用各个模块 */
func PostLocalMsgWithVal(channel string, val interface{}) {
	for i, mc := range __ntfcenter.chanels {
		if mc.Msg == channel {
			for _, sub := range __ntfcenter.chanels[i].Sublsit {
				sub.OnMsgNotifyWithVal(channel, val)
			}
		}
	}
}
/*
listenPubSubChannels listens for messages on Redis pubsub channels. The
--- onStart function is called after the channels are subscribed.
--- onMessage function is called for each message.
*/
func SetupWithRedis(host, port, pwd string) error {
	redisServerAddr := fmt.Sprintf("%s:%s", host, port)
	// A ping is set to the server with this period to test for the health of the connection and server.
	const healthCheckPeriod = time.Minute

	c, err := redis.Dial("tcp",
		redisServerAddr,
		redis.DialPassword(pwd),
		redis.DialReadTimeout(healthCheckPeriod+10*time.Second), // Read timeout on server should be greater than ping period.
		redis.DialWriteTimeout(10*time.Second))

	if err != nil {
		lego.LogError(err)
		return err
	}
	defer c.Close()
	psc := redis.PubSubConn{Conn: c}

	channels := make([]string, 0)
	for _, mc := range __ntfcenter.chanels {
		channels = append(channels, mc.Msg)
	}

	if err := psc.Subscribe(redis.Args{}.AddFlat(channels)...); err != nil {
		lego.LogError(err)
		return err
	}
	defer psc.Unsubscribe()

	done := make(chan error, 1)
	// Start a goroutine to receive notifications from the server.
	go func() {
		for {
			switch n := psc.Receive().(type) {
			case error:
				continue
			case redis.Message:
				for i, mc := range __ntfcenter.chanels {
					if mc.Msg == n.Channel {
						for _, sub := range __ntfcenter.chanels[i].Sublsit {
							sub.OnMsgNotify(n.Channel, string(n.Data))
						}
					}
				}
			case redis.Subscription:
				continue
			}
		}
	}()

	ticker := time.NewTicker(healthCheckPeriod)
	defer ticker.Stop()

	for err == nil {
		select {
		case <-ticker.C:
			// Send ping to test health of connection and server. If
			// corresponding pong is not received, then receive on the
			// connection will timeout and the receive goroutine will exit.
			if err = psc.Ping(""); err != nil {
				lego.LogError(err)
				psc.Unsubscribe()
				// 重新连接
				psc.Subscribe(redis.Args{}.AddFlat(channels)...)
				//goto loop
			}
		case err := <-done:
			// Return error from the receive goroutine.
			return err
		}
	}
	return nil
}
